package cn.itcast.tags.models.ml

import cn.itcast.tags.models.{AbstractModel, ModelType}
import cn.itcast.tags.tools.{MLModelTools, TagTools}
import org.apache.spark.ml.{Pipeline, PipelineModel}
import org.apache.spark.ml.classification.{DecisionTreeClassificationModel, DecisionTreeClassifier}
import org.apache.spark.ml.evaluation.MulticlassClassificationEvaluator
import org.apache.spark.ml.feature.{VectorAssembler, VectorIndexer, VectorIndexerModel}
import org.apache.spark.ml.param.ParamMap
import org.apache.spark.ml.tuning.{ParamGridBuilder, TrainValidationSplit, TrainValidationSplitModel}
import org.apache.spark.sql.expressions.UserDefinedFunction
import org.apache.spark.sql.functions._
import org.apache.spark.sql.{Column, DataFrame, SparkSession}

/**
 * 挖掘类型标签:用户购物性别USG模型，使用管道模型
 */
class UsgTagModel extends AbstractModel("用户购物性别USG模型", ModelType.ML) {
  /*
    378	用户购物性别
      379	男		0
      380	女		1
      381	中性		-1
   */
  override def doTag(businessDF: DataFrame, tagDF: DataFrame): DataFrame = {
    val spark: SparkSession = businessDF.sparkSession
    import spark.implicits._

    // businessDF.printSchema()
    // businessDF.show(10, truncate = false)
    // tagDF.printSchema()
    // tagDF.filter($"level" === 5).show(10, truncate = false)

    // 1. 获取订单表数据tbl_orders,与订单商品表数据关联获取会员ID
    val ordersDF: DataFrame = spark.read
      .format("hbase")
      .option("zkHosts", "bigdata-cdh01.itcast.cn")
      .option("zkPort", "2181")
      .option("hbaseTable", "tbl_tag_orders")
      .option("family", "detail")
      .option("selectFields", "memberid,ordersn")
      .load()
    // ordersDF.printSchema()
    // ordersDF.show(30, truncate = false)

    // 2. 加载维度数据:tbl_dim_colors(颜色)/tbl_dim_products(产品)
    // 2.1 加载颜色维度表数据
    val colorsDF: DataFrame = spark.read
      .format("jdbc")
      .option("driver", "com.mysql.jdbc.Driver")
      .option("url", "jdbc:mysql://bigdata-cdh01.itcast.cn:3306/?useUnicode=true&characterEncoding=UTF-8&serverTimezone=UTC")
      .option("dbtable", "profile_tags.tbl_dim_colors")
      .option("user", "root")
      .option("password", "123456")
      .load()
    // colorsDF.printSchema()
    // colorsDF.show(30, truncate = false)

    // 2.2 构建颜色WHEN语句
    val colorColumn: Column = {
      // 声明变量
      var colorCol: Column = null
      colorsDF
        .as[(Int, String)].rdd
        .collectAsMap()
        .foreach { case (colorId, colorName) =>
          if (null == colorCol) {
            colorCol = when($"ogcolor".equalTo(colorName), colorId)
          } else {
            colorCol = colorCol.when($"ogcolor".equalTo(colorName), colorId)
          }
        }
      colorCol = colorCol.otherwise(0).as("color")
      // 返回
      colorCol
    }

    // 2.3 加载商品维度表数据
    val productsDF: DataFrame = spark.read
      .format("jdbc")
      .option("drvier", "com.mysql.jdbc.Driver")
      .option("url", "jdbc:mysql://bigdata-cdh01.itcast.cn:3306/?useUnicode=true&characterEncoding=UTF-8&serverTimezone=UTC")
      .option("dbtable", "profile_tags.tbl_dim_products")
      .option("user", "root")
      .option("password", "123456")
      .load()
    // productsDF.printSchema()
    // productsDF.show(30, truncate = false)

    // 2.4. 构建颜色WHEN语句
    val productColumn: Column = {
      // 声明变量
      var productCol: Column = null
      productsDF
        .as[(Int, String)].rdd
        .collectAsMap()
        .foreach { case (productId, productName) =>
          if (null == productCol) {
            productCol = when($"producttype".equalTo(productName),
              productId)
          } else {
            productCol =
              productCol.when($"producttype".equalTo(productName), productId)
          }
        }
      productCol = productCol.otherwise(0).as("product")
      // 返回
      productCol
    }

    // 2.5. 根据运营规则标注的部分数据
    val labelColumn: Column = {
      when($"ogcolor".equalTo("樱花粉")
        .or($"ogcolor".equalTo("白色"))
        .or($"ogcolor".equalTo("香槟色"))
        .or($"ogcolor".equalTo("香槟金"))
        .or($"productType".equalTo("料理机"))
        .or($"productType".equalTo("挂烫机"))
        .or($"productType".equalTo("吸尘器/除螨仪")), 1) // 女
        .otherwise(0) // 男
        .alias("label") // 决策树预测label
    }

    // 3. 商品数据和订单数据关联
    val goodsDF: DataFrame = businessDF
      // 依据订单号ordersn关联
      .join(ordersDF, businessDF("cordersn") === ordersDF("ordersn"))
      // 将颜色、产品使用ID替换
      .select(
        $"memberid".as("userId"),
        colorColumn, productColumn, labelColumn
      )
    // goodsDF.printSchema()
    // goodsDF.show(50, truncate = false)

    // TODO: 构建算法模型，并使用模型预测，管道pipeline模型
    // val pipelineModel: PipelineModel = trainPipelineModel(goodsDF)
    // val predictionDF: DataFrame = pipelineModel.transform(goodsDF)

    // TODO: 使用训练验证分割方式，设置超参数训练获取最佳模型
    // val pipelineModel: PipelineModel = trainBestModel(goodsDF)
    // val dtcModel = pipelineModel.stages(2).asInstanceOf[DecisionTreeClassificationModel]
    // println(dtcModel.toDebugString)
    // val predictionDF: DataFrame = pipelineModel.transform(goodsDF)

    // TODO: 使用交叉训练方式，设置超参数训练获取最佳模型
    val pipelineModel = MLModelTools.loadModel(goodsDF, "usg", this.getClass)
    val predictionDF: DataFrame = pipelineModel.transform(goodsDF)

    //predictionDF.printSchema()
    //predictionDF.show(100, truncate = false)
    // 4. 按照用户ID分组，统计每个用户购物，男女商品个数
    val genderDF: DataFrame = predictionDF
      .groupBy($"userId")
      .agg(
        count($"userId").as("total"), // 计算用户所有商品数目
        // 计算用户所有商品中属于男性数目
        sum(
          when($"label".equalTo(0), 1).otherwise(0)
        ).as("maleTotal"),
        // 计算用户所有商品中属于女性数目
        sum(
          when($"label".equalTo(1), 1).otherwise(0)
        ).as("femaleTotal")
      )
    // genderDF.printSchema()
    // genderDF.show(10, truncate = false)

    // 5. 计算标签，自定义UDF函数
    // 5.1 属性标签数据中规则rule
    val rulesMap: Map[String, String] = TagTools.convertMap(tagDF)
    val rulesMapBroadcast = spark.sparkContext.broadcast(rulesMap)

    // 5.2 计算每个用户，近半年订单中的男性商品超过60%认定为男，或近半年女性商品超过60%认定为女
    val gender_tag_udf: UserDefinedFunction = udf(
      (total: Long, maleCount: Double, femaleCount: Double) => {
        val maleRate: Double = maleCount / total
        val femaleRate: Double = femaleCount / total
        if (maleRate >= 0.6) {
          rulesMapBroadcast.value("0")
        } else if (femaleRate >= 0.6) {
          rulesMapBroadcast.value("1")
        } else {
          rulesMapBroadcast.value("-1")
        }
      }
    )

    // 5.3 根据该用户的total/maleCount/femaleCount判断用户应该被打上哪个标签
    val modelDF: DataFrame = genderDF.select(
      $"userId",
      gender_tag_udf($"total", $"maleTotal", $"femaleTotal").as("usg")
    )
    // modelDF.printSchema()
    // modelDF.show(100, truncate = false)

    modelDF
  }

  /**
   * 调整算法超参数，找出最优模型
   * @param dataframe 数据集
   * @return
   */
  def trainBestModel(dataframe: DataFrame): PipelineModel = {
    // a. 特征向量化
    val assembler: VectorAssembler = new VectorAssembler()
      .setInputCols(Array("color", "product"))
      .setOutputCol("raw_features")
    // b. 类别特征进行索引
    val indexer: VectorIndexer= new VectorIndexer()
      .setInputCol("raw_features")
      .setOutputCol("features")
      .setMaxCategories(30)
    // c. 构建决策树分类器
    val dtc: DecisionTreeClassifier = new DecisionTreeClassifier()
      .setFeaturesCol("features")
      .setLabelCol("label")
      .setPredictionCol("prediction")
    // d. 构建Pipeline管道流实例对象
    val pipeline: Pipeline = new Pipeline().setStages(
      Array(assembler, indexer, dtc)
    )
    // TODO: e. 构建参数网格，设置超参数的值
    val paramGrid: Array[ParamMap] = new ParamGridBuilder()
      .addGrid(dtc.maxDepth, Array(5, 10))
      .addGrid(dtc.maxBins, Array(32, 64))
      .addGrid(dtc.impurity, Array("gini", "entropy"))
      .build()
    // f. 多分类评估器
    val evaluator = new MulticlassClassificationEvaluator()
      .setLabelCol("label")
      .setPredictionCol("prediction")
      // 指标名称，支持：f1、weightedPrecision、weightedRecall、accuracy
      .setMetricName("accuracy")
    // TODO: g. 训练验证
    val trainValidationSplit = new TrainValidationSplit()
      .setEstimator(pipeline)
      .setEvaluator(evaluator)
      .setEstimatorParamMaps(paramGrid)
    // 80% 的数据将用于训练，其余 20% 用于验证。
      .setTrainRatio(0.8)
    // h. 训练模型
    val model: TrainValidationSplitModel = trainValidationSplit.fit(dataframe)
    // i. 获取最佳模型返回
    model.bestModel.asInstanceOf[PipelineModel]
  }

  /**
   * 使用决策树分类算法训练模型，返回PipelineModel模型
   */
  def trainPipelineModel(dataframe: DataFrame): PipelineModel = {
    // 数据划分为训练数据集和测试数据集
    val Array(trainingDF, testingDF) = dataframe.randomSplit(Array(0.8, 0.2), seed = 123)
    // a. 特征向量化
    val assembler: VectorAssembler = new VectorAssembler()
      .setInputCols(Array("color", "product"))
      .setOutputCol("raw_features")
    // b. 类别特征进行索引
    val indexer: VectorIndexer = new VectorIndexer()
      .setInputCol("raw_features")
      .setOutputCol("features")
      .setMaxCategories(30)
    // c. 构建决策树分类器
    val dtc: DecisionTreeClassifier = new DecisionTreeClassifier()
      .setFeaturesCol("features")
      .setLabelCol("label")
      .setPredictionCol("prediction")
      .setImpurity("gini") // 基尼系数
      .setMaxDepth(5) // 树的深度
      .setMaxBins(32) // 树的叶子数目
    // d. 构建Pipeline管道流实例对象
    val pipeline: Pipeline = new Pipeline().setStages(
      Array(assembler, indexer, dtc)
    )
    // e. 训练模型
    val pipelineModel: PipelineModel = pipeline.fit(trainingDF)
    // f. 模型评估
    val predictionDF: DataFrame = pipelineModel.transform(testingDF)
    predictionDF.show(100, truncate = false)
    println(s"accuracy = ${modelEvaluate(predictionDF, "accuracy")}")
    // 返回模型
    pipelineModel
  }

  /**
   * 模型评估，返回计算分类指标值
   *
   * @param dataframe  预测结果的数据集
   * @param metricName 分类评估指标名称，支持：f1、weightedPrecision、weightedRecall、accuracy
   */
  def modelEvaluate(dataframe: DataFrame, metricName: String): Double = {
    // a. 构建多分类分类器
    val evaluator = new MulticlassClassificationEvaluator()
      .setLabelCol("label")
      .setPredictionCol("prediction")
      // 指标名称，
      .setMetricName(metricName)
    // b. 计算评估指标
    val metric: Double = evaluator.evaluate(dataframe)
    // c. 返回指标
    metric
  }
}

object UsgTagModel {
  def main(args: Array[String]): Unit = {
    val tagModel = new UsgTagModel()
    tagModel.executeModel(378L)
  }
}

